/*
    Copyright 2007-2008 Adrien Guillon.  All Rights Reserved.

    This file is part of TBB Community Code. 

    TBB Community Code is free software; you can redistribute it
    and/or modify it under the terms of the GNU General Public License
    version 2 as published by the Free Software Foundation.

    TBB Community Code is distributed in the hope that it will be
    useful, but WITHOUT ANY WARRANTY; without even the implied warranty
    of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
    along with TBB Community Code; if not, write to the Free Software
    Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA

    As a special exception, you may use this file as part of a free software
    library without restriction.  Specifically, if other files instantiate
    templates or use macros or inline functions from this file, or you compile
    this file and link it with other files to produce an executable, this
    file does not by itself cause the resulting executable to be covered by
    the GNU General Public License.  This exception does not however
    invalidate any other reasons why the executable file might be covered by
    the GNU General Public License.
*/

#ifndef __TCC_parallel_for_group_H
#define __TCC_parallel_for_group_H

#include "separator.h"
#include "grouped_range.h"
#include "ungrouped_range.h"


// Use internal parallel_for which supports non-const Range
#include "internal/nonconst_parallel_for.h"

#include "tbb/task.h"
#include "tbb/partitioner.h"

namespace tcc { // START NAMESPACE tcc


namespace internal { // START NAMESPACE internal

/**
 * Helper class that simply returns a task that executes
 * a parallel_for. 
 */
template <typename Range, typename Body, typename Partitioner>
class do_parallel_for : public tbb::task
{
public:
    do_parallel_for(Range& range, const Body& body, const Partitioner& partitioner)
    : _range(range), _body(body), _partitioner(partitioner)
    { }

    tbb::task* execute()
    {
        // Currently use the internal parallel for, which supports
        // writes to a range
        tcc::internal::parallel_for(_range, _body, _partitioner);
        return 0;
    }

private:
    Range& _range;
    const Body& _body;
    const Partitioner& _partitioner;
};


/**
 * This is a helper class to start the parallel processing of grouped
 * and ungrouped objects, and it is the main logic which drives
 * parallel_for_group.
 */
template<typename SeparationType, typename Body, typename Partitioner>
class start_for_group : public tbb::task
{
public:

    start_for_group(SeparationType& separation, const Body& body, const Partitioner& partitioner) :
    _separation(separation),
    _body(body),
    _partitioner(partitioner)
    { }

    tbb::task* execute()
    {

        typedef grouped_range<typename SeparationType::grouped_container_type> grouped_range_type;
        typedef ungrouped_range<typename SeparationType::ungrouped_container_type> ungrouped_range_type;

        // Create TBB ranges for grouped and ungrouped objects
        grouped_range_type range_of_grouped(_separation.grouped);
        ungrouped_range_type range_of_ungrouped(_separation.ungrouped);

        typedef do_parallel_for<grouped_range_type, Body, Partitioner> for_grouped_type;
        typedef do_parallel_for<ungrouped_range_type, Body, Partitioner> for_ungrouped_type;


        // Fork two tasks, one to process groups and one to process ungrouped items

        for_grouped_type& process_grouped = *new( allocate_child() ) for_grouped_type(range_of_grouped, _body, _partitioner);
        for_ungrouped_type& process_ungrouped = *new( allocate_child() ) for_ungrouped_type(range_of_ungrouped, _body, _partitioner);

        // Reference count for two children and the wait
        set_ref_count(3);

        // Spawn children and wait
        spawn(process_grouped);
        spawn_and_wait_for_all(process_ungrouped);

        return 0;
    }

private:
    SeparationType& _separation;
    const Body& _body;
    const Partitioner& _partitioner;
};


} // END NAMESPACE internal


/**
 * parallel_for_group
 *
 * Provides parallel iteration over grouped and ungrouped parts of the 
 * result of the operation of a separator on a composite object.  So basically
 * an object is broken apart into groups and ungrouped objects, and each
 * will be iterated over in parallel.
 *
 * The separation that was used for computation is returned.
 * 
 * TODO: This might not be the best way in terms of performance, since
 * a copy of the separation_type is going to be made and it could be
 * quite large.
 *
 * The SeparatorType must publicly inherit separator, and provide the following:
 *   separation separate(const CompositeType&)
 *
 *   (not required but recommended)
 *   bool isSeparable(const CompositeType&)
 *
 * The Body must provide the following:
 *   void operator()(SeparatorType::grouped_range_type& x)
 *   void operator()(SeperatorType::ungrouped_range_type& x)
 *
 */
template <typename CompositeType, typename SeparatorType, typename Body, typename Partitioner>
typename SeparatorType::separation_type parallel_for_group(const CompositeType& composite, const SeparatorType& separator, const Body& body, const Partitioner& partitioner)
{
    typedef typename SeparatorType::separation_type separation_type;
    typedef internal::start_for_group<separation_type, Body, Partitioner> start_for_type;

    separation_type separated = separator.separate(composite);

    // We decide how to continue computation based upon the results
    // of the separation.  We only process ungrouped / grouped if
    // they are non-empty, and both if they are both non-empty.
    // This is done to avoid the overhead of task creation.

    if(!separated.grouped.empty() && !separated.ungrouped.empty())
    {
        // Construct a task class which will fork two children, one to
        // iterate in parallel over grouped items, and the other to iterate in 
        // parallel over ungrouped objects.

        start_for_type& root = *new( tbb::task::allocate_root() ) start_for_type(separated, body, partitioner);
        tbb::task::spawn_root_and_wait(root);
    }
    else if (!separated.grouped.empty())
    {
        typedef grouped_range<typename SeparatorType::separation_type::grouped_container_type> grouped_range_type;
        grouped_range_type range_of_grouped(separated.grouped);
        tcc::internal::parallel_for(range_of_grouped, body, partitioner);
    }
    else if(!separated.ungrouped.empty())
    {
        typedef ungrouped_range<typename SeparatorType::separation_type::ungrouped_container_type> ungrouped_range_type;
        ungrouped_range_type range_of_ungrouped(separated.ungrouped);
        tcc::internal::parallel_for(range_of_ungrouped, body, partitioner);
    }

    // If both grouped/ungrouped are empty, we do nothing but return the separation.

    return separated;
}


} // END NAMESPACE tcc

#endif





